package queue

import (
	"cashcenter/utils/logx"
	"context"
	"github.com/pkg/errors"
	"github.com/spf13/viper"
	"github.com/streadway/amqp"
	"sync"
)

var conn *amqp.Connection
var (
	MQChan *amqp.Channel
	// ...
)
var (
	Xcg    string
	pre    = "ctpQueueCash"
	RkWait = "Wait"
	RkExec = "Exec"
	RkDead = "Dead"
)

func MqInit(ctx context.Context, sysWaitGroup *sync.WaitGroup) {
	sysWaitGroup.Add(1)

	var err error

	url := viper.GetString("rabbitmq.url")
	Xcg = viper.GetString("rabbitmq.exchange")

	conn, err = amqp.Dial(url)
	if err != nil {
		logx.LogrusObj.Fatalln(errors.Wrap(err, "[QUEUE] connect fail."))
	}
	defer conn.Close()
	MQChan, err = conn.Channel()
	if err != nil {
		logx.LogrusObj.Fatalln(errors.Wrap(err, "[QUEUE] open channel fail."))
	}
	defer MQChan.Close()

	// === 区分逻辑点 ===
	err = xAndQ(Xcg)
	// 开启失败，尝试删除
	if err != nil {
		var count int
		var errDel error
		count, errDel = MQChan.QueueDelete(RkDead, false, false, false)
		if errDel != nil {
			if count != 0 {
				err = errors.Wrap(
					err,
					(errors.Wrapf(errDel, "Queue delete failed, there left %d message.", count)).Error(),
				)
			} else {
				err = errors.Wrap(err, errDel.Error())
			}
		}
		count, errDel = MQChan.QueueDelete(RkExec, false, false, false)
		if errDel != nil {
			if count != 0 {
				err = errors.Wrap(
					err,
					(errors.Wrapf(errDel, "Queue delete failed, there left %d message.", count)).Error(),
				)
			} else {
				err = errors.Wrap(err, errDel.Error())
			}
		}
		count, errDel = MQChan.QueueDelete(RkWait, false, false, false)
		if errDel != nil {
			if count != 0 {
				err = errors.Wrap(
					err,
					(errors.Wrapf(errDel, "Queue delete failed, there left %d message.", count)).Error(),
				)
			} else {
				err = errors.Wrap(err, errDel.Error())
			}
		}
		errDel = MQChan.ExchangeDelete(Xcg, false, false)
		if errDel != nil {
			err = errors.Wrap(
				err,
				(errors.Wrap(errDel, "Exchange delete failed.")).Error(),
			)
		}
		logx.LogrusObj.Fatalln(errors.Wrap(err, "[QUEUE]"))
	}

	// 开启消费
	consume(Xcg)

	<-ctx.Done()

	logx.LogrusObj.Println("[QUEUE] connect done.")
	sysWaitGroup.Done()
}

func xAndQ(exchange string) error {
	var err error

	// 声明 exchange
	err = MQChan.ExchangeDeclare(
		exchange, //name
		"direct",
		true,
		false,
		false,
		false,
		nil,
	)
	if err != nil {
		return errors.Wrap(err, "Exchange declare failed")
	}
	// 声明 queueWait
	queueWait, err := MQChan.QueueDeclare(
		pre+RkWait, // name
		true,
		false,
		false,
		false,
		amqp.Table{
			"x-message-ttl":             60000,
			"x-dead-letter-exchange":    exchange,
			"x-dead-letter-routing-key": RkExec,
			"x-max-length":              10240,
			"x-queue-mode":              "lazy",
		},
	)
	if err != nil {
		return errors.Wrap(err, "Queue declare failed")
	}
	// 声明 queueExec
	queueExec, err := MQChan.QueueDeclare(
		pre+RkExec, // name
		true,
		false,
		false,
		false,
		amqp.Table{
			"x-dead-letter-exchange":    exchange,
			"x-dead-letter-routing-key": RkDead,
			"x-max-length":              10240,
			"x-queue-mode":              "lazy",
		},
	)
	if err != nil {
		return errors.Wrap(err, "Queue declare failed")
	}
	// 声明 queueDead
	queueDead, err := MQChan.QueueDeclare(
		pre+RkDead, // name
		true,
		false,
		false,
		false,
		amqp.Table{
			"x-dead-letter-exchange": exchange,
			"x-max-length":           10240,
			"x-queue-mode":           "lazy",
		},
	)
	if err != nil {
		return errors.Wrap(err, "Queue declare failed")
	}

	// exchange 绑定 queues
	err = MQChan.QueueBind(queueWait.Name, RkWait, exchange, false, nil)
	if err != nil {
		return errors.Wrap(err, "Queue bind failed")
	}
	err = MQChan.QueueBind(queueExec.Name, RkExec, exchange, false, nil)
	if err != nil {
		return errors.Wrap(err, "Queue bind failed")
	}
	err = MQChan.QueueBind(queueDead.Name, RkDead, exchange, false, nil)
	if err != nil {
		return errors.Wrap(err, "Queue bind failed")
	}

	return nil
}
